package com.mango.ch09;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.mango.HDFSTools.HDFSAPI;

/**
 * 将mapper和reduce写在一个文件中的编写方式
 * 
 * 文本文件 ：GroomFriendJob.txt
 * 
 * @author Mango
 *
 */
public class FriendRecommendJob extends Configured implements Tool {
	static Path inputPath = new Path("/FriendRecommendJob/input");// input
	static Path outPath = new Path("/FriendRecommendJob/output"); // output

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new Configuration(), new FriendRecommendJob(), args);
		System.exit(res);
	}

	// job0负责将购物清单按物品进行分组
	public static class FRMapper extends Mapper<LongWritable, Text, LongWritable, MyTuple> {
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] splits = value.toString().split(" ");
			Integer userKey = Integer.valueOf(splits[0]);
			List<Long> friends = iterableToList(splits[1].split(","));
			// 输出直接好友列表、、、用于在后面的reduce中进行比较某两个用户ID 是否是直接好哟关系
			for (Long fid : friends) {
				// -1表示两用户之间是直接好友关系
				context.write(new LongWritable(userKey), new MyTuple(fid, -1));
			}
			// 发出所有可能的好友关系(即设置相互好友，列出所有情况如 F有好友a,b,c, 则context写出 a,(b,F) 和b,(a,F)
			// c,(a,F)，a,(c,F)，b,(c,F)，c,(b,F)的所有情况)
			int count = 0;
			for (int i = 0; i < friends.size(); i++)
				for (int j = i + 1; j < friends.size(); j++) {
					// 可能1
					MyTuple mt1 = new MyTuple(friends.get(j), userKey);
					context.write(new LongWritable(friends.get(i)), mt1);
					// 可能2
					MyTuple mt2 = new MyTuple(friends.get(i), userKey);
					context.write(new LongWritable(friends.get(j)), mt2);
					count += 1;
				}
			System.out.println("总共的可能好友关系有" + count + "种");
		}
	}

	public static class FRReducer extends Reducer<LongWritable, MyTuple, Text, Text> {
		@Override
		protected void reduce(LongWritable currenUser, Iterable<MyTuple> values, Context context)
				throws IOException, InterruptedException {
			// mutualFriends.key 是推荐的好友
			// mutualFriends.value是共同好友列表
			Map<Long, List<Integer>> mutualFriends = new HashMap<>();
			for (MyTuple myTuple : values) {// values中是推荐用户列表
				Long toUser = myTuple.get_1();// 推荐用户的ID
				Integer mutualFriend = myTuple.get_2();// 当前用户与该推荐用户的共同好友
				boolean isFriend = (mutualFriend == -1);// 判断当前用户与该推荐用户的好友关系是不是直接好友
				System.out.println(
						"当前用户" + currenUser + " 推荐用户 " + toUser + " 关系 :" + ((isFriend) ? "直接好友" : myTuple.get_2()));
				// 当前用户3 推荐用户 8 关系 :1
				// else---添加到共同好友8之前
				// 当前用户3 推荐用户 7 关系 :1
				// else---添加到共同好友7之前 推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 4 关系 :1
				// else---添加到共同好友4之前 推荐人:7共同好友列表:[1]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 5 关系 :1
				// else---添加到共同好友5之前 推荐人:4共同好友列表:[1]推荐人:7共同好友列表:[1]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 2 关系 :直接好友
				// 当前用户3 推荐用户 1 关系 :直接好友
				// 当前用户3 推荐用户 6 关系 :1
				// else---添加到共同好友6之前
				// 推荐人:4共同好友列表:[1]推荐人:5共同好友列表:[1]推荐人:7共同好友列表:[1]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 7 关系 :2
				// if---添加共同好友7之前
				// 推荐人:4共同好友列表:[1]推荐人:5共同好友列表:[1]推荐人:6共同好友列表:[1]推荐人:7共同好友列表:[1]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 5 关系 :2
				// if---添加共同好友5之前 推荐人:4共同好友列表:[1]推荐人:5共同好友列表:[1]推荐人:6共同好友列表:[1]推荐人:7共同好友列表:[1,
				// 2]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 4 关系 :2
				// if---添加共同好友4之前 推荐人:4共同好友列表:[1]推荐人:5共同好友列表:[1,
				// 2]推荐人:6共同好友列表:[1]推荐人:7共同好友列表:[1, 2]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 1 关系 :2
				// else---添加到共同好友1之前 推荐人:4共同好友列表:[1, 2]推荐人:5共同好友列表:[1,
				// 2]推荐人:6共同好友列表:[1]推荐人:7共同好友列表:[1, 2]推荐人:8共同好友列表:[1]
				// 当前用户3 推荐用户 2 关系 :1

				if (mutualFriends.containsKey(toUser))// 判断该推荐好友是否已经在存在
				{// 若存在，则判断是不是和用户是直接好友关系
					if (isFriend)
					// 若是好友关系，则不用管，让其共同好友列表为null
					{
						mutualFriends.put(toUser, null);// 必须要将此key存入map且将value其置null，
						// 否则当第一次遇到(3,(1,-1))判断是直接好友不会存入，而在第二次(3,(1,2))时 会判断非直接好友会存入，
						// 所以第一次时就一定要存入，且置null，第二次遇到时判断null即可，时null表示时直接好友
					} else if (mutualFriends.get(toUser) != null) {
						// 否则，将该推荐用户添加到推荐好友列表中
						System.out.println("if---添加共同好友" + toUser + "之前 " + mapToString(mutualFriends));
						mutualFriends.get(toUser).add(mutualFriend);
					}
				} else {
					if (isFriend)// 判断是否是直接好友关系,若是，将其共同好友列表置为null
					{
						mutualFriends.put(toUser, null);
					} else
					// 添加进共同好友集合中
					{
						System.out.println("else---添加到共同好友" + toUser + "之前 " + mapToString(mutualFriends));
						List<Integer> list = new ArrayList<>();
						list.add(mutualFriend);
						mutualFriends.put(toUser, list);
					}

				}
			}
			context.write(new Text(currenUser.get() + ""), new Text(buildOutput(mutualFriends)));

		}
	}

	@Override
	public int run(String[] args) throws Exception {
		// Configuration processed by ToolRunner
		// 在执行APP之前检查路径问题
		checkPath();
		Configuration conf = getConf();
		Job job = Job.getInstance(conf);
		job.setJarByClass(this.getClass());
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outPath);
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(MyTuple.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(3);
		job.setMapperClass(FriendRecommendJob.FRMapper.class);
		job.setReducerClass(FriendRecommendJob.FRReducer.class);

		// .....
		// 构建job1
		// Job job1 = Job.getInstance(conf, "job1");
		// ......

		// 各个job间的顺序控制
		ControlledJob cj1 = new ControlledJob(conf);
		cj1.setJob(job);
		// ControlledJob cj2 = new ControlledJob(conf);
		// job间的依赖关系
		// cj2.addDependingJob(cj1);
		JobControl jc = new JobControl(this.getClass().getSimpleName());
		// 执行全部job （按顺序)
		jc.addJob(cj1);
		new Thread(jc).start();
		while (true) {
			for (ControlledJob cj : jc.getRunningJobList()) {
				cj.getJob().monitorAndPrintJob();

			}
			if (jc.allFinished())
				break;
		}
		// 第二种执行job的方法
		// Submit the job, then poll for progress until the job is complete
		/*
		 * RunningJob runningJob = JobClient.runJob(job); if (runningJob.isSuccessful())
		 * { return 0; } else { return 1; } }
		 */

		return 0;
	}

	protected static List<Long> iterableToList(String[] strings) {
		// TODO Auto-generated method stub
		List<Long> list = new ArrayList<>();
		for (String it : strings) {
			list.add(Long.valueOf(it));
		}
		return list;
	}

	private static String buildOutput(Map<Long, List<Integer>> mutualFriends) {
		// TODO Auto-generated method stub
		String output = "";
		for (Entry<Long, List<Integer>> entry : mutualFriends.entrySet()) {
			List<Integer> v = entry.getValue();
			String k = entry.getKey() + "";
			if (v != null)
				output += k + "(" + v.size() + ":" + v.toString() + ") ";
		}
		return output;
	}

	private static String mapToString(Map<Long, List<Integer>> mutualFriends) {
		StringBuilder sb = new StringBuilder();
		for (Entry<Long, List<Integer>> item : mutualFriends.entrySet()) {
			sb.append("推荐人:");
			sb.append(item.getKey().toString());
			sb.append("共同好友列表:");
			List<Integer> list = item.getValue();
			if (list != null)
				sb.append(list.toString());
			else
				sb.append("[]");
		}
		return sb.toString();
	}

	public static void checkPath() {
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			hdfs.createDirectory(inputPath);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			hdfs.closeFilseSystem();
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
